#pragma once

#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"

#include "../mq_comm/mq_helper.hpp"
#include "../mq_comm/mq_log.hpp"
#include "../mq_comm/mq_msg.pb.h"
#include "../mq_comm/mq_threadpool.hpp"
#include "../mq_comm/mq.pb.h"

#include "mq_consumer.hpp"

#include <iostream>
#include <mutex>
#include <condition_variable>
#include <unordered_map>

namespace mq
{
    using namespace ns_helper;

    typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec):
        _cid(UUIDHelper::uuid()), _conn(conn), _codec(codec) {}
        ~Channel()
        {
            // 如果是消费者, 需要取消订阅
            // 这里不指定取消订阅也可以, 服务端会自动释放信道并取消订阅
            basicCancel();
        }

        std::string cid() { return _cid; }

        // 打开信道
        bool openChannel()
        {
            std::string rid = UUIDHelper::uuid();
            openChannelRequest req; 
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn, req);
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }
        // 关闭信道
        void closeChannel()
        {
            std::string rid = UUIDHelper::uuid();
            closeChannelRequest req; 
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn, req);
            waitResponse(rid);
            return ;
        }

        // 声明交换机
        bool declareExchange(
            const std::string &name,
            ExchangeType type, 
            bool durable, 
            bool auto_delete,
            google::protobuf::Map<std::string, std::string> &args)
        {
            // 构建请求
            declareExchangeRequest req;
            std::string rid =  UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            req.set_exchange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);

            // 向服务器发送请求
            _codec->send(_conn, req);
            //等待服务器的响应
            basicCommonResponsePtr resp = waitResponse(rid);
            // 返回
            return resp->ok();
        }

        // 删除交换机
        void deleteExchange(const std::string &name)
        {
            std::string rid = UUIDHelper::uuid();
            deleteExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            _codec->send(_conn, req);
            waitResponse(rid);
            return ;
        }

        // 声明队列
        bool declareQueue(
            const std::string &qname, 
            bool qdurable, 
            bool qexclusive,
            bool qauto_delete,
            google::protobuf::Map<std::string, std::string> &qargs)
        {
            std::string rid = UUIDHelper::uuid();
            declareQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_durable(qdurable);
            req.set_auto_delete(qauto_delete);
            req.set_exclusive(qexclusive);
            req.mutable_args()->swap(qargs);
            _codec->send(_conn, req);
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }
        // 删除队列
        void deleteQueue(const std::string &qname)
        {
            std::string rid = UUIDHelper::uuid();
            deleteQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            _codec->send(_conn, req);
            waitResponse(rid);
            return ;
        }

        // 交换机,绑定队列
        bool queueBind(
            const std::string &ename, 
            const std::string &qname, 
            const std::string &key)
        {
            std::string rid = UUIDHelper::uuid();
            queueBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_binding_key(key);
            _codec->send(_conn, req);
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }

        // 解绑队列(交换机解除绑定)
        void queueUnBind(const std::string &ename, const std::string &qname)
        {
            std::string rid = UUIDHelper::uuid();
            queueUnBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            _codec->send(_conn, req);
            waitResponse(rid);
            return ;
        }
        // 消息发布
        void basicPublish(
            const std::string &ename,
            const BasicProperties *bp,
            const std::string &body) 
        {
            std::string rid = UUIDHelper::uuid();

            basicPublishRequest req;

            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_body(body);
            req.set_exchange_name(ename);
            if (bp != nullptr) {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, req);
            waitResponse(rid);
            return ;
        }
        // 消息确认
        void basicAck(const std::string &msgid)
        {
            if (_consumer.get() == nullptr) {
                LOG(ERROR) << "消息确认时，找不到消费者信息！\n";
                return ;
            }
            std::string rid = UUIDHelper::uuid();
            basicAckRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_message_id(msgid);
            _codec->send(_conn, req);
            waitResponse(rid);
            return;
        }
        // 取消订阅
        void basicCancel()
        {
            if (_consumer.get() == nullptr) 
            {
                return ;
            }
            std::string rid = UUIDHelper::uuid();
            basicCancelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_consumer_tag(_consumer->tag);
            _codec->send(_conn, req);
            waitResponse(rid);
            return;
        }
        // 添加订阅
        bool basicConsume(
            const std::string &consumer_tag,
            const std::string &queue_name,
            bool auto_ack,
            const ConsumerCallBack &cb)
        {
            if (_consumer.get() != nullptr) 
            {
                LOG(ERROR) << "当前信道已订阅其他队列消息！\n";
                return false;
            }
            std::string rid = UUIDHelper::uuid();
            basicConsumeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(queue_name);
            req.set_consumer_tag(consumer_tag);
            req.set_auto_ack(auto_ack);
            _codec->send(_conn, req);
            basicCommonResponsePtr resp =  waitResponse(rid);
            if (resp->ok() == false) 
            {
                LOG(ERROR) << "添加订阅失败！\n";
                return false;
            }
            _consumer = std::make_shared<Consumer>(consumer_tag, queue_name, auto_ack, cb);
            return true;
        }
    private:
        basicCommonResponsePtr waitResponse(const std::string &rid)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            // 条件满足就唤醒
            _cv.wait(lock, [this, &rid](){
                return _basic_resp.find(rid) != _basic_resp.end();
            });

            basicCommonResponsePtr basic_resp = _basic_resp[rid];
            _basic_resp.erase(rid);
            return basic_resp;
        }
    public:
        //连接收到基础响应后，向hash_map中添加响应
        void putBasicResponse(const basicCommonResponsePtr& resp) 
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _basic_resp.insert(std::make_pair(resp->rid(), resp));
            _cv.notify_all();
        }

        //连接收到消息推送后，需要通过信道找到对应的消费者对象，通过回调函数进行消息处理
        // 客户端业务处理的回调函数
        void consume(const basicConsumeResponsePtr& resp) 
        {
            if (_consumer.get() == nullptr) {
                LOG(ERROR) << "消息处理时，未找到订阅者信息！\n";
                return;
            }
            if (_consumer->tag != resp->consumer_tag()) {
                LOG(ERROR) << "收到的推送消息中的消费者标识，与当前信道消费者标识不一致！\n";
                return ;
            }
            // 调用回调函数进行消息处理
            _consumer->callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
        }
    private:
        std::string _cid;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        Consumer::ptr _consumer; // 一个信道对应一个消费者
        std::mutex _mtx;

        std::condition_variable _cv;
        std::unordered_map<std::string, basicCommonResponsePtr> _basic_resp; // 请求id 与 响应的映射

    };
    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager()
        {}

        Channel::ptr create(const muduo::net::TcpConnectionPtr &conn, 
                const ProtobufCodecPtr &codec)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            auto channel = std::make_shared<Channel>(conn, codec);
            _channels.insert(std::make_pair(channel->cid(), channel));
            return channel;
        }
        void remove(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _channels.erase(cid);
        }
        Channel::ptr get(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mtx);

            auto it = _channels.find(cid);
            if (it == _channels.end()) {
                return Channel::ptr();
            }
            return it->second;
        }
    private:
        std::mutex _mtx;
        std::unordered_map<std::string ,Channel::ptr> _channels; // 信道id 与 信道的映射
    };
}

